package com.tul.vapp.servicecenter.client;

import com.tul.cusa.common.util.ListUtil;
import com.tul.cusa.zookeeper.IChildListener;
import com.tul.cusa.zookeeper.INodeListener;
import com.tul.cusa.zookeeper.IZookeeperClient;
import com.tul.vapp.servicecenter.common.ZkNodeConstant;
import com.tul.vapp.servicecenter.utils.ConfigUtil;
import com.tul.vapp.servicecenter.zookeeper.ZooKeeperClientFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 配置仓库，包含从ZK获取的配置和从文件中加载的配置
 * 
 * @author wwxiang
 * @since 2017/12/28.
 */
public class Repository {

	private static final Logger LOGGER = LoggerFactory.getLogger(Repository.class);

	private static final String ROOT = "/config";
	private static final String APP_GLOBAL = "global";

	private static final String SEPARATOR = "/";

	private static final String SERVICE_CENTER = "service-center";

	private boolean startRemote;

	private String appId;

	private IZookeeperClient client;

	/**
	 * Map<namespace, Map<key, value>>
	 */
	private Map<String, Map<String, String>> localRepo = new ConcurrentHashMap<>();
	private Map<String, Map<String, String>> zkRepo = new ConcurrentHashMap<>();

	/**
	 * 构造函数
	 * 
	 * @param appId 应用ID
	 * @param startRemote 是否开启从远端获取配置
	 */
	public Repository(String appId, boolean startRemote) {
		this.startRemote = startRemote;
		this.appId = appId;
		if (startRemote) {
			this.client = ZooKeeperClientFactory.getClient();
		}
	}

	/**
	 * subscribe
	 */
	public void subscribe() {
		if (!startRemote) {
			LOGGER.warn("Repository doesn't start remote config.");
			return;
		}
		subscribe(appId);
		subscribe(APP_GLOBAL);
	}

	/**
	 * 1.关注命名空间节点的个数<br/>
	 * 2.关注各个空间的内容
	 * 
	 * @param appNode app节点
	 */
	public void subscribe(String appNode) {
		// 确保service-center已启动
		blockingUntilSCEnable();
		// 确保APP节点存在
		String appPath = ROOT + SEPARATOR + appNode;
		if (!client.exists(appPath)) {
			client.create(appPath, false);
		}
		LOGGER.info("Subscribe app path " + appPath);

		// 监听APP节点
		List<String> namespaces = client.addChildListener(appPath, new IChildListener() {
			@Override
			public void childChanged(String path, List<String> children) {

				List<String> current = new ArrayList(zkRepo.keySet());
				List<String> add = ListUtil.subtract(children, current);
				List<String> delete = ListUtil.subtract(current, children);

				for (String ns : add) {
					LOGGER.info("Add node " + path + SEPARATOR + ns);
					// 添加节点后,新增监听;
					String initContent = client.addNodeListener(appPath + SEPARATOR + ns, new NodeListenerImpl());
					// 初始化对应空间的zkRepo, 此方法做过同步处理
					handleContentChange(ns, initContent);
				}

				for (String ns : delete) {
					LOGGER.info("Delete node " + path + SEPARATOR + ns);
					// 移除zkRepo中的配置
					zkRepo.remove(ns);
				}
			}
		});

		// 监听namespace节点内容
		for (String namespace : namespaces) {
			// 新增监听;
			String initContent = client.addNodeListener(appPath + SEPARATOR + namespace, new NodeListenerImpl());
			// 初始化对应空间的zkRepo, 此方法做过同步处理
			handleContentChange(namespace, initContent);
		}
	}

	private void blockingUntilSCEnable() {
		// SC的状态节点如果存在，说明SC已启动完毕
		boolean started = scEnable();
		while (!started) {
			LOGGER.warn("Waiting for service-center starting. Check in 3 seconds later.");
			try {
				TimeUnit.SECONDS.sleep(3);
			} catch (InterruptedException e) {
				LOGGER.error("Interrupted when waiting for service-center starting.", e);
			}
			started = scEnable();
		}
	}

	/**
	 * 只要/runtime节点下有节点包含service-center 就说明SC已启动
	 * 如/runtime/service-center@10.168.18.50
	 * 
	 * @return service-center是否成功启动
	 */
	private boolean scEnable() {
		List<String> availableApps = client.getChildren(ZkNodeConstant.STATUS_NODE);
		if (availableApps == null) {
			return false;
		}
		for (String app : availableApps) {
			if (app.contains(SERVICE_CENTER)) {
				return true;
			}
		}
		return false;
	}

	private class NodeListenerImpl implements INodeListener {

		@Override
		public void nodeChanged(String path, String content) {
			int idx = path.lastIndexOf(SEPARATOR);
			String namespace = path.substring(idx + 1);
			handleContentChange(namespace, content);
		}
	}

	private synchronized void handleContentChange(String namespace, String content) {
	    // 此方法会在主线程里调用，事件线程也会调用，因此做同步处理
		LOGGER.info(namespace + " content changed.");
		try {
			Properties props = ConfigUtil.parse(content);
			Map<String, String> contentMap = ConfigUtil.format(props);
			zkRepo.put(namespace, contentMap);
		} catch (IOException e) {
			LOGGER.error("Content change exception", e);
		}
	}

	/**
	 * 将本地文件加载到仓库中
	 * 
	 * @param namespace 命名空间
	 * @param properties 配置props
	 */
	public void loadFile(String namespace, Properties properties) {
		Map<String, String> content = ConfigUtil.format(properties);
		localRepo.put(namespace, content);
	}

	/**
	 * 获取配置 后续用下面两个方法替代
	 * 
	 * @see #getPropertyInLocal(String, String)
	 * @see #getPropertyInZk(String, String)
	 * 
	 * @param namespace 命名空间
	 * @param key 配置key
	 * @return 配置值
	 */
	@Deprecated
	public String getProperty(String namespace, String key) {
		String result = null;

		// 从ZK获取
		Map<String, String> zkProps = zkRepo.get(namespace);
		if (zkProps != null) {
			result = zkProps.get(key);
		}
		// 获取到值后直接返回
		if (result != null) {
			return result;
		}

		// 从本地获取
		Map<String, String> localProps = localRepo.get(namespace);
		if (localProps != null) {
			result = localProps.get(key);
			if (result != null) {
                //LOGGER.warn("Config from local " + key + "=" + result);
            }
		}
		return result;
	}

	/**
	 * 根据命名空间获取配置项
	 * 
	 * @param namespace 命名空间
	 * @return 配置项
	 */
	public Map<String, String> getProperties(String namespace) {
		Map<String, String> result = new HashMap();
		Map<String, String> localProps = localRepo.get(namespace);
		if (localProps != null) {
			result.putAll(localProps);
		}
		Map<String, String> zkProps = zkRepo.get(namespace);
		if (zkProps != null) {
			result.putAll(zkProps);
		}
		return result;
	}

	/**
	 * 列出此仓库中所有的命名空间
	 * 
	 * @return 此仓库中所有的命名空间
	 */
	public Set<String> listNamespace() {
		Set<String> ns = new LinkedHashSet();
		ns.addAll(zkRepo.keySet());
		ns.addAll(localRepo.keySet());
		return ns;
	}

	/**
	 * 获取配置
	 *
	 * @param namespace 命名空间
	 * @param key 配置key
	 * @return 配置值
	 */
	public String getPropertyInZk(String namespace, String key) {
		String result = null;
		// 从ZK获取
		Map<String, String> zkProps = zkRepo.get(namespace);
		if (zkProps != null) {
			result = zkProps.get(key);
		}
		return result;
	}

	/**
	 * 获取配置
	 *
	 * @param namespace 命名空间
	 * @param key 配置key
	 * @return 配置值
	 */
	public String getPropertyInLocal(String namespace, String key) {
		String result = null;
		// 从本地获取
		Map<String, String> localProps = localRepo.get(namespace);
		if (localProps != null) {
			result = localProps.get(key);
			if (result != null) {
				LOGGER.warn("Config from local " + key + "=" + result);
			}
		}
		return result;
	}
}
